package com.lizhiyu.flink.demo5_window;


import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Random;


/***
 * 并行度为多个时候，open方法执行多次 的自定义数据源
 */
public class CustomSource3 extends RichParallelSourceFunction<String> {
    private List<String> title = Arrays.asList("java","javaee","mysql","redis","ES","kafka","rocketmq","rabbitmq","dubbo","zk");
    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        while (true){
            Thread.sleep(1000L);
            Random random = new Random();
            int randomNum = random.nextInt(10);
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("hh-mm-ss");
            sourceContext.collect(title.get(randomNum)+":"+randomNum+":"+simpleDateFormat.format(new Date()));
        }
    }

    @Override
    public void cancel() {
        System.out.println("执行关闭方法");
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("执行初始化方法、并发度有几个就执行几次open方法...");
    }


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();
        evn.setParallelism(2);
        evn.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        DataStreamSource<String> dataStreamSource = evn.addSource(new CustomSource3());
        dataStreamSource.print(Thread.currentThread().getName());
        evn.execute();
    }
}
